package com.yitiao.quickstart;

import com.yitiao.constants.Const;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

public class Producer {
    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("test_quick_producer_name");
        // 设置 NameServerAdd
        producer.setNamesrvAddr(Const.NAMESRV_ADDR_MASTER_SLAVE);

        // 生产者参数配置
//        producer.setRetryTimesWhenSendFailed();
//        producer.setSendLatencyFaultEnable(true);  // 开启自动容错机制
        producer.start();

        for (int i = 30; i < 40; i++) {
            Message message = new Message(Const.TEST_QUICKSTART_TOPIC,    //	主题
                    "TagA", //	标签
                    "key-" + i,    // 	用户自定义的key ,唯一的标识
                    ("Hello RocketMQ" + i).getBytes()
            );
            // 延迟投递
//            message.setDelayTimeLevel(4);
            // 同步发送
            SendResult result = producer.send(message);
            // 指定队列发送
//            SendResult result = producer.send(message, new MessageQueueSelector() {
//                @Override
//                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
//                    return mqs.get((Integer) arg);
//                }
//            },1);
            System.out.println(result);
            // 异步发送
//            producer.send(message, new SendCallback() {
//                @Override
//                public void onSuccess(SendResult sendResult) {
//                    System.err.println("msgId: " + sendResult.getMsgId() + ", status: " + sendResult.getSendStatus());
//                }
//
//                @Override
//                public void onException(Throwable e) {
//                    e.printStackTrace();
//                    System.err.println("------发送失败");
//                }
//            });

        }
        producer.shutdown();
    }

}
